package rocketmq

import (
	"time"

	"testkjcld.bhecard.com/gitlab/kit/esim/rocketmq/apache"

	"testkjcld.bhecard.com/gitlab/kit/esim/rocketmq/aliyun"
	"testkjcld.bhecard.com/gitlab/kit/esim/rocketmq/internal"

	"testkjcld.bhecard.com/gitlab/kit/esim/config"
	"testkjcld.bhecard.com/gitlab/kit/esim/log"
)

type Publisher struct {
	client internal.RocketProxy

	logger log.Logger

	conf config.Config
}

type PublisherOption func(*Publisher)

func NewPublisher(options ...PublisherOption) *Publisher {
	p := &Publisher{}

	for _, option := range options {
		option(p)
	}

	if p.conf == nil {
		p.conf = config.NewNullConfig()
	}

	if p.logger == nil {
		p.logger = log.NewLogger()
	}

	rocketMq := p.conf.GetString("rocket_mq_type")
	if rocketMq == internal.RocketMqApache {
		/*apache 初始化*/
		var cliOpt apache.MQClientOptions
		p.client = apache.NewMQClient(
			cliOpt.WithLogger(p.logger),
			cliOpt.WithConf(p.conf))
	} else {
		/*aliyun 版本初始化*/
		var cliOpt aliyun.MQClientOptions
		p.client = aliyun.NewMQClient(
			cliOpt.WithLogger(p.logger),
			cliOpt.WithConf(p.conf))
	}

	return p
}

func WithPublisherConf(conf config.Config) PublisherOption {
	return func(p *Publisher) {
		p.conf = conf
	}
}

func WithPublisherLogger(logger log.Logger) PublisherOption {
	return func(p *Publisher) {
		p.logger = logger
	}
}

// PublishMessage with message.
func (p *Publisher) PublishMessage(topicName, messageBody string) error {
	return p.client.PublishMessage(topicName, messageBody)
}

// PublishMsgWithTag with message tag and key.
func (p *Publisher) PublishMsgWithTag(topicName, messageBody, messageTag string) error {
	return p.client.PublishMsgWithTag(topicName, messageBody, messageTag)
}

// PublishMsgWithKeyTag with message tag.
func (p *Publisher) PublishMsgWithKeyTag(topicName, messageBody, messageTag, messageKey string) error {
	return p.client.PublishMsgWithKeyTag(topicName, messageBody, messageTag, messageKey)
}

func (p *Publisher) PublishMessageProp(topicName, messageTag, messageBody string, properties map[string]string) error {
	return p.client.PublishMessageProp(topicName, messageTag, messageBody, properties)
}

func (p *Publisher) PublishDelayMessage(topicName, messageBody string, delay time.Duration) error {
	return p.client.PublishDelayMessage(topicName, messageBody, delay)
}

func (p *Publisher) PublishDelayMsgWithTag(topicName, messageBody, messageTag string, delay time.Duration) error {
	return p.client.PublishDelayMsgWithTag(topicName, messageBody, messageTag, delay)
}

func (p *Publisher) PublishDelayMsgWithKeyTag(topicName, messageBody, messageTag, messageKey string, delay time.Duration) error {
	return p.client.PublishDelayMsgWithKeyTag(topicName, messageBody, messageTag, messageKey, delay)
}

func (p *Publisher) PublishDelayMessageProt(topicName, messageBody string, properties map[string]string, delay time.Duration) error {
	return p.client.PublishDelayMessageProt(topicName, messageBody, properties, delay)
}

// PublishMsgWithTagTrans with message tag and group trans.
func (p *Publisher) PublishMsgWithTagTrans(topicName, groupID, messageBody, messageTag string, tradeID map[string]string) (*internal.ConsumeMessageAck, error) {
	return p.client.PublishMsgWithTagTrans(topicName, groupID, messageBody, messageTag, tradeID)
}

// PublishWithCommit trans commit.
func (p *Publisher) PublishWithCommit(topicName, groupID, handler string) error {
	return p.client.PublishWithCommit(topicName, groupID, handler)
}

// PublishWithRollBack trans rollback.
func (p *Publisher) PublishWithRollBack(topicName, groupID, handler string) error {
	return p.client.PublishWithRollBack(topicName, groupID, handler)
}
